package server_agent

import (
	"lark/conf"
	"lark/consts"
	"lark/proto"
	"net"
	"strings"
	"sync/atomic"
	"time"

	"lark/helper"
	"lark/util"
)

//接收来自 client 的 task，转发到业务 server 并返回结果
type serverConn struct {
	id        string
	taskChan  chan *TaskReqMsg
	parse     *proto.LarkProtoV2
	svc       *Server
	closeChan chan struct{}
	status    uint32
	nettype   string
	netaddr   string
	kind      string
	usedAt    time.Time
}

func newServerConn(id string, ch chan *TaskReqMsg, addr string, kind string, svc *Server) *serverConn {
	var c = &serverConn{
		id:        id,
		taskChan:  ch,
		svc:       svc,
		status:    newFlag,
		kind:      kind,
		closeChan: make(chan struct{}),
	}

	n := strings.Split(addr, "://")
	if len(n) == 1 {
		c.nettype = "unix"
		c.netaddr = addr
	} else {
		c.nettype = n[0]
		c.netaddr = n[1]
	}

	return c
}

func (c *serverConn) dial() bool {
	if atomic.LoadUint32(&c.status) == closedFlag {
		return false
	}

	if c.parse != nil {
		c.parse.Close()
	}

	conn, err := net.DialTimeout(c.nettype, c.netaddr, conf.ServerAgentDialLocalTimeout())
	if err != nil {
		logger.Errorf("[%s] dial %s %s fail. %s", c.id, c.nettype, c.netaddr, err)
		return false
	}

	c.parse = proto.NewlarkProtoV2(conn, conf.ServerAgentReadLocalTimeout(), conf.ServerAgentWriteLocalTimeout())
	c.usedAt = time.Now()
	logger.Debugf("[%s] dial %s %s success.", c.id, c.nettype, c.netaddr)

	return true
}

func (c *serverConn) run() {
	defer util.PrintPanicStack()
	defer c.close()

	c.dial()

	for {
		select {
		case task := <-c.taskChan:
			if task == nil {
				return
			}

			var msgid = task.Req.GetId()
			logger.Debugf("[%s] receive task. msgid:%d", c.id, msgid)

			var resp *proto.TaskResp
			if time.Since(task.CreatedAt) > conf.ServerAgentRequestMaxWaitingTime() {
				resp = &proto.TaskResp{
					Id:     msgid,
					Status: consts.ServerAgentWaitingTimeout,
				}

			} else {
				resp = c.handleTask(task)
				if resp.Status != consts.ServerAgentOK {
					c.dial()
				}
			}

			select {
			case task.Client.WriteChan() <- resp:
				logger.Debugf("[%s] write message success. msgid:%d", c.id, msgid)

			default:
				select {
				case task.Client.WriteChan() <- resp:
					logger.Debugf("[%s] write message success. msgid:%d", c.id, msgid)

				case <-time.After(time.Millisecond * 100):
					logger.Errorf("[%s] write message failed, drop. msgid:%d", c.id, msgid)
				}
			}

		case <-c.closeChan:
			return
		}
	}
}

func (c *serverConn) handleTask(task *TaskReqMsg) *proto.TaskResp {
	msgid := task.Req.GetId()
	reqdata := task.Req.Data
	var tag uint16
	var payload []byte
	var err error

	if time.Since(c.usedAt) > conf.ServerAgentToBusinessMaxIdleTime {
		c.dial()
	} else {
		c.usedAt = time.Now()
	}

	//send data to business server and get response
	for tryNum := 2; tryNum > 0; tryNum-- {
		err = c.parse.Write(consts.Lsp2psTagTaskReq, reqdata, helper.JsonMarshal(map[string]string{
			"client_ip": task.ClientIP,
		}))
		if err != nil {
			if tryNum > 0 && c.dial() {
				logger.Warnf("[%s] write data failed, retry. msgid:%d, error:%s, data:%q", c.id, msgid, err, reqdata)
				continue
			} else {
				logger.Errorf("[%s] write data failed, drop. msgid:%d, error:%s, data:%q", c.id, msgid, err, reqdata)
				return &proto.TaskResp{
					Id:     task.Req.GetId(),
					Status: consts.ServerAgentWriteBusinessServerError,
				}
			}
		}
		logger.Debugf("[%s] write data success. msgid:%d, data:%q", c.id, msgid, task.Req.GetData())

		tag, payload, err = c.parse.Read()
		if err != nil {
			if c.kind == RegisterKindPHP && tryNum > 0 && isConnresetErr(err) && c.dial() {
				logger.Warnf("[%s] read data failed, retry. msgid:%d, error:%s, data:%q", c.id, msgid, err, reqdata)
				continue
			} else {
				logger.Errorf("[%s] read data failed, drop. msgid:%d, error:%s, data:%q", c.id, msgid, err, reqdata)
				return &proto.TaskResp{
					Id:     task.Req.GetId(),
					Status: consts.ServerAgentReadBusinessServerError,
				}
			}
		}

		if logger.LogDebug() {
			logger.Debugf("[%s] read data success. msgid:%d, data:%q", c.id, msgid, reqdata)
		}

		break
	}

	//check the tag
	if tag != consts.Lsp2psTagTaskRespOK {
		logger.Errorf("[%s] receive wrong tag, want:%d, get:%d", c.id, consts.Lsp2psTagTaskRespOK, tag)
		return &proto.TaskResp{
			Id:     task.Req.GetId(),
			Status: consts.ServerAgentInternalError,
		}
	}

	return &proto.TaskResp{
		Id:     task.Req.GetId(),
		Status: consts.ServerAgentOK,
		Data:   proto.SplitPayload(payload)[0],
	}
}

func (c *serverConn) closed() bool {
	return atomic.LoadUint32(&c.status) == closedFlag
}

func (c *serverConn) close() {
	if atomic.CompareAndSwapUint32(&c.status, runningFlag, closedFlag) {
		if c.parse != nil {
			c.parse.Close()
		}

		close(c.closeChan)

		c.svc.DelServerConn(c.id)
	}
}

func isConnresetErr(err error) bool {
	if err == nil {
		return false
	}

	operr, ok := err.(*net.OpError)
	if !ok {
		return false
	}

	//if operr.Err.Error() == syscall.ECONNRESET.Error() {
	if operr.Err.Error() == "read: connection reset by peer" {
		return true
	}

	return false
}
